package com.it.state;

import com.it.operator.utils.SourceUtils;
import com.it.pojo.Event;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * 关于状态的测试
 *
 * @author code1997
 */
public class StateTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        executionEnvironment.setParallelism(1);
        SingleOutputStreamOperator<Event> eventStream = SourceUtils.getEventSource(executionEnvironment).assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO).withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
            @Override
            public long extractTimestamp(Event element, long recordTimestamp) {
                return element.timestamp;
            }
        }));
        eventStream.keyBy(data -> data.user)
                .flatMap(new MyFlatMapFunction()).print();
        executionEnvironment.execute();
    }

    public static class MyFlatMapFunction extends RichFlatMapFunction<Event, String> {

        ValueState<Event> myValueState;
        ListState<Event> myListState;
        MapState<String, Long> myMapState;
        ReducingState<Long> myReducingState;
        AggregatingState<Event, String> myAggregatingState;
        Long count;


        @Override
        public void open(Configuration parameters) throws Exception {
            //状态的初始化放到算子的生命周期方法中
            ValueStateDescriptor<Event> myValueState = new ValueStateDescriptor<>("myValueState", Types.POJO(Event.class));
            this.myValueState = getRuntimeContext().getState(myValueState);
            myListState = getRuntimeContext().getListState(new ListStateDescriptor<Event>("myListState", Types.POJO(Event.class)));
            myMapState = getRuntimeContext().getMapState(new MapStateDescriptor<String, Long>("myMapState", Types.STRING, Types.LONG));
            myReducingState = getRuntimeContext().getReducingState(new ReducingStateDescriptor<Long>("myReducingState", new ReduceFunction<Long>() {
                @Override
                public Long reduce(Long value1, Long value2) throws Exception {
                    return value1 + value2;
                }
            }, Types.LONG));
            myAggregatingState = getRuntimeContext().getAggregatingState(new AggregatingStateDescriptor<Event, String, String>("myAggregatingState", new AggregateFunction<Event, String, String>() {
                @Override
                public String createAccumulator() {
                    return "url的集合：\n";
                }

                @Override
                public String add(Event value, String accumulator) {
                    return accumulator + "\n" + value.url;
                }

                @Override
                public String getResult(String accumulator) {
                    return accumulator;
                }

                @Override
                public String merge(String a, String b) {
                    return null;
                }
            }, Types.STRING));
            //本地变量相较于算子状态缺少了故障恢复的处理.
            count = 0L;

            StateTtlConfig stateTtlConfig = StateTtlConfig.newBuilder(Time.hours(1)).build();
            myValueState.enableTimeToLive(stateTtlConfig);

        }

        @Override
        public void flatMap(Event value, Collector<String> out) throws Exception {
            //value
            //System.out.println(myValueState.value());
            myValueState.update(value);
            //System.out.println("==update==" + myValueState.value());
            //list
            myListState.add(value);
            //map
            Long tempCount = myMapState.get(value.user);
            myMapState.put(value.user, (tempCount == null) ? 1 : (tempCount + 1));
            System.out.println("==myMapState::" + value.user + "==" + myMapState.get(value.user));
            //myReducingState
            myReducingState.add(value.timestamp);
            System.out.println("==myReducingState==" + myReducingState.get());
            //myAggregatingState
            myAggregatingState.add(value);
            System.out.println("==myAggregatingState==" + myAggregatingState.get());
            count++;
            System.out.println("==count==" + count);

        }
    }
}
